/**
 *
 */
package org.cmcc.pcap.jpcap.tcp;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import org.cmcc.pcap.jpcap.util.PacketAnalysisConfig;

/**
 * @Date 2018年1月17日 上午8:38:12
 * @author Amish
  * 保存一组Connection及其对应的Packet，以connectionKey区分
 * TODO 统计信息单独放到TcpConnectionCacheStatistics类中
 */
public class TcpConnectionCache{
	public static Logger logger = LogManager.getLogger(TcpConnectionCache.class);
	
	/**
	 *  执行完成时
	 * totalTcpPacketNum = receiveInvalidTcpPacketNum + overTimeInvalidTcpPacketNum + pickupPacketNum
	 * pickupPacketNum = handledPacketNum
	 * 所以
	 * totalTcpPacketNum - (receiveInvalidTcpPacketNum + overTimeInvalidTcpPacketNum + handledPacketNum) = 0 才表示所有包处理完成
	 */
	public AtomicInteger totalTcpPacketNum;//缓冲区收到的所有TCP Packet数量
	public AtomicInteger receiveInvalidTcpPacketNum;//收到的所有无效TCP Packet数量（无对应TCP初始消息的）//TODO 无效协议的应该也放到这里？
	public AtomicInteger overTimeInvalidTcpPacketNum;//超时无效并remove掉的TCP Packet数量 （无对应TCP结束消息的）
	public AtomicInteger pickupPacketNum;//缓冲区已经提取出的TCP Packet数量
	public AtomicInteger handledPacketNum;//被工作线程/消费者线程 “处理完”的TCP Packet数量
	
	public AtomicInteger totalConnectionNum;//进入到cache的Connection总数
	public AtomicInteger finishConnectionNum;//处理完的Connection总数，包括正常处理的、没有结束消息删除的和无效协议删除的等
	
	//ConcurrentHashMap<String ,TcpConnection> chm;//TODO 是否考虑换成性能更好的数据结构
	LinkedHashMap <String ,TcpConnection> chm;//用LinkedHashMap替换ConcurrentHashMap，保证遍历的顺序性
	
	long overtime = 25000;//TCP连接处理的超时时间，超过该时间没有收到FIN/RST包，该连接认为是非法的，直接删除Connection
	long delaytime = 5000;//TCP连接处理的延迟时间，收到FIN或RST后，延迟一定时间处理，以保证后续相关包能放入Connection
	int addthreshold = 20;//向TcpConnectionCache增加pakcet时，生产者阻塞的条件，这里的值表示cache中TcpConnection的数量
	int getthreshold = 6;//从TcpConnectionCache读取TcaConnection时，消费者阻塞的条件，这里的值表示已经读取的符合条件的TcpConnection的数量
	int tcpPacketsThreshold = 10000;//超过该值的tcp connection不做后续处理
	
	public TcpConnectionCache(){
		//chm = new ConcurrentHashMap<String ,TcpConnection>();
		chm = new LinkedHashMap <String ,TcpConnection>();
		totalTcpPacketNum = new AtomicInteger(0);
		receiveInvalidTcpPacketNum = new AtomicInteger(0);
		overTimeInvalidTcpPacketNum = new AtomicInteger(0);
		pickupPacketNum = new AtomicInteger(0);
		handledPacketNum =  new AtomicInteger(0);
		totalConnectionNum = new AtomicInteger(0);
		finishConnectionNum = new AtomicInteger(0);
		
		overtime = PacketAnalysisConfig.getConfigAsLong("tcp.connection.overtime");
		delaytime = PacketAnalysisConfig.getConfigAsLong("tcp.connection.delaytime");
		addthreshold = PacketAnalysisConfig.getConfigAsInt("tcp.connection.cache.addthreshold");
		getthreshold = PacketAnalysisConfig.getConfigAsInt("tcp.connection.cache.getthreshold");
		tcpPacketsThreshold = PacketAnalysisConfig.getConfigAsInt("tcp.connection.packets.threshold");
	}
	
	/**
	 * 此处会过滤掉没有三次握手消息第一条SYN消息的TcpConnection对应的所有Tcp Packet
	 * @param tcpPacketDecorator
	 */
	public synchronized void addTcpPacketDecorator(TcpPacketDecorator tcpPacketDecorator){
		
		
		this.totalTcpPacketNum.getAndIncrement();
		while( chm.size()>=addthreshold ){//TODO 如何更合理的选择这个控制条件？
			try {
				this.wait();
			} catch (InterruptedException e) {
				logger.error(e.getCause(),e);
			}
		}
		//每收到一个TCP包，生成4元组对应的一个Key（注意消息方向），唯一标识一个TCP连接
		String src_ip = tcpPacketDecorator.tcpPacket.src_ip.getHostAddress();
		String src_port = Integer.toString(tcpPacketDecorator.tcpPacket.src_port);
		String dst_ip = tcpPacketDecorator.tcpPacket.dst_ip.getHostAddress();
		String dst_port =  Integer.toString(tcpPacketDecorator.tcpPacket.dst_port);
		String src = src_ip+":"+src_port;
		String dst = dst_ip+":"+dst_port;
		String connectionKey = src+"-"+dst;       //通过四元组生成connectionKey
		String connectionKeyReverse = dst+"-"+src;//通过四元组生成connectionKey，源目的对调
		
		
		/**
		 * 如果是初始消息
		 *     *检查集合里是否包含connectionKey对应的TcpConnection对象
		 *         如果没有，新建TcpConnection对象，并保存消息及相关信息
		 *         如果有，并保存消息及相关信息(三次握手的第一条消息也可能重传)
		 * 如果是不是初始消息
		 *     *检查集合里是否包含connectionKey对应的TcpConnection对象
		 *         如果没有对应的TcpConnection，扔掉该消息
		 *         如果有对应的TcpConnection，
		 *             *判断这个是应用层协议是否需要处理
		 *                 需要处理，则保存消息及相关信息
		 *                 不需要处理，直接删除对应的TcpConnection
		 */
		TcpConnection tcpConnection = null;
		try{
			if(tcpPacketDecorator.isFirst){//tcp 三次握手第一条消息，该消息可能有重传消息
				tcpConnection = chm.get(connectionKey);
				
				if(tcpConnection == null ){
					tcpConnection = new TcpConnection();
					tcpConnection.connectionKey = connectionKey;
					tcpConnection.clientAddr = src;
					tcpConnection.serverAddr = dst;
					tcpConnection.al.add(tcpPacketDecorator);
					tcpConnection.lastPacketTime = tcpPacketDecorator.receiveTime;//更新connection最后一个packet的加入cache的时间(注:非抓包时间)
					chm.put(connectionKey, tcpConnection);
					this.totalConnectionNum.incrementAndGet();
				}else{
					tcpConnection.al.add(tcpPacketDecorator);
					tcpConnection.lastPacketTime = tcpPacketDecorator.receiveTime;//更新connection最后一个packet的处理时间(注:非抓包时间)
				}
				
			}else{
				tcpConnection = chm.get(connectionKey) !=null? chm.get(connectionKey) : chm.get(connectionKeyReverse);
				if(tcpConnection == null){
					receiveInvalidTcpPacketNum.getAndIncrement();
					logger.debug("can not find the match connectionKey with :"+connectionKey+" or "+connectionKeyReverse);
				}else{
					
					if(tcpConnection.firstDataTcpPacketDecorator == null && tcpPacketDecorator.tcpPacket.data.length > 0){
						/**
						 * 判断应用层协议类型,无法判断时(含HTTP后续分段时)/data长度为零时  都会返回null
						 * 未在配置列表里时会返回NOTDEAL
						 */
						TcpProtocol tcpProtocol = TcpProtocolFilter.filterTcpProtocol(tcpPacketDecorator) ;
						
						if(tcpProtocol != null && tcpProtocol != TcpProtocol.NOTDEAL){
							//tcpPacketDecorator.protocol = tcpProtocol;//设置应用层协议，如HTTP。在无法判断时也可能是TCP
							tcpConnection.firstDataTcpPacketDecorator = tcpPacketDecorator;//给connection设置第一个携带数据的包
							tcpConnection.protocol = tcpProtocol;//给connection设置已识别的应用层协议
	
							tcpConnection.al.add(tcpPacketDecorator);
							tcpConnection.haveFin = tcpPacketDecorator.tcpPacket.fin ==true ? tcpPacketDecorator.tcpPacket.fin : tcpConnection.haveFin;//有一个fin消息即更新
							tcpConnection.haveRST = tcpPacketDecorator.tcpPacket.rst ==true ? tcpPacketDecorator.tcpPacket.rst : tcpConnection.haveRST;//有一个rst消息即更新
							tcpConnection.lastPacketTime = tcpPacketDecorator.receiveTime;//更新connection最后一个packet的处理时间(注:非抓包时间)
							//logger.info("add  a  tcpPacketDecorator:"+tcpConnection.haveFin+" // "+tcpConnection.haveRST+" // "+tcpConnection.lastPacketTime);
						}else{//第一个带data的包即无法识别的Protocol，直接删除connection   //TODO 这里忽略了一个包乱序的极小概率事件
							//chm.remove(connectionKey);
							chm.remove(tcpConnection.connectionKey);
							this.finishConnectionNum.incrementAndGet();
							this.receiveInvalidTcpPacketNum.getAndAdd(tcpConnection.al.size()+1);//加1是因为当前这个包本身也要属于非法包
						}
					}else{
						tcpConnection.al.add(tcpPacketDecorator);
						tcpConnection.haveFin = tcpPacketDecorator.tcpPacket.fin ==true ? tcpPacketDecorator.tcpPacket.fin : tcpConnection.haveFin;//有一个fin消息即更新
						tcpConnection.haveRST = tcpPacketDecorator.tcpPacket.rst ==true ? tcpPacketDecorator.tcpPacket.rst : tcpConnection.haveRST;//有一个rst消息即更新
						tcpConnection.lastPacketTime = tcpPacketDecorator.receiveTime;//更新connection最后一个packet的处理时间(注:非抓包时间)
						//logger.info("add  a  tcpPacketDecorator:"+tcpConnection.haveFin+" // "+tcpConnection.haveRST+" // "+tcpConnection.lastPacketTime);
					}				
				}
			}		
		
		}catch(Exception e){
			logger.error("Some exception when add packet to cache : ", e);
			if(tcpConnection != null)
			    logger.error(tcpConnection.connectionKey);
			receiveInvalidTcpPacketNum.getAndIncrement();
		}
		this.notify();
	}
	
	public synchronized ArrayList<TcpConnection> getTcpConnection(){
		ArrayList<TcpConnection> al = new ArrayList<TcpConnection>();//TODO 传入一个？
		try{
		
			while(chm.values().size() == 0){//TODO 如何选择这个控制条件？
				try {
					this.wait();
				} catch (InterruptedException e) {
					logger.error(e.getCause(),e);
				}
			}
			
			
			int i=getthreshold;//每次处理多少个合法TCP连接？
			
			Iterator<Entry<String,TcpConnection>> iterator = chm.entrySet().iterator();
			while(iterator.hasNext()) {
			//for( Entry<String, TcpConnection> entry : chm.entrySet()){
				if(i==0) {
					this.notify();
					return al;
				}
					
				TcpConnection tcpConnection = iterator.next().getValue();//entry.getValue();
				
				
	// 20180419
	//			if((System.currentTimeMillis() - tcpConnection.lastPacketTime) > overtime){//注意，是判断最后一条消息距离当前时间，是不是超过overtime值，TOFIX 这个判断可能与处理性能有关系
	//				if(!tcpConnection.haveFin && !tcpConnection.haveRST){//超时，且没有tcp结束事件，扔掉....
	//					chm.remove(tcpConnection.connectionKey);
	//					overTimeInvalidTcpPacketNum.getAndAdd(tcpConnection.al.size());
	//					logger.info("overtime and hava not finish or reset message,remove this connection and its packet :"+tcpConnection.connectionKey);
	//					logger.info("remove connection<"+tcpConnection.connectionKey+"> and its packets : "+tcpConnection.al.size());
	//				}else{
	//					al.add(tcpConnection);
	//					chm.remove(tcpConnection.connectionKey);
	//					processedPacketNum.getAndAdd(tcpConnection.al.size());
	//				}			
	//			}else{
	//				//do nothing
	//			}
				
				if(tcpPacketsThreshold!=0 && tcpConnection.al.size() > tcpPacketsThreshold){//包数超过阈值的链接直接删除20180813
					iterator.remove();//chm.remove(tcpConnection.connectionKey);
					this.finishConnectionNum.incrementAndGet();
					this.receiveInvalidTcpPacketNum.getAndAdd(tcpConnection.al.size());
					logger.info("too lang tcp connection,remove this connection and its packet :"+tcpConnection.connectionKey);
					logger.info("remove connection<"+tcpConnection.connectionKey+"> and its packets : "+tcpConnection.al.size());
				}else{
					if(!tcpConnection.haveFin && !tcpConnection.haveRST){//没有tcp结束事件
						if((System.currentTimeMillis() - tcpConnection.lastPacketTime) > overtime){//注意，是判断最后一条消息距离当前时间，是不是超过overtime值，TOFIX 这个判断可能与处理性能有关系
							iterator.remove();//chm.remove(tcpConnection.connectionKey);
							this.finishConnectionNum.incrementAndGet();
							overTimeInvalidTcpPacketNum.getAndAdd(tcpConnection.al.size());
							logger.info("overtime and hava not finish or reset message,remove this connection and its packet :"+tcpConnection.connectionKey);
							logger.info("remove connection<"+tcpConnection.connectionKey+"> and its packets : "+tcpConnection.al.size());
							i--;
						}else{
							//logger.info("*******************************wait for stop FIN or RST packet");
						}
					}else {
						if((System.currentTimeMillis() - tcpConnection.lastPacketTime) > delaytime){//注意，是判断最后一条消息距离当前时间，是不是超过overtime值，TOFIX 这个判断可能与处理性能有关系
							al.add(tcpConnection);
							iterator.remove();//chm.remove(tcpConnection.connectionKey);
							this.finishConnectionNum.incrementAndGet();
							pickupPacketNum.getAndAdd(tcpConnection.al.size());
							
							i--;//放在这里执行是为了保证处理够i设定的数量才释放锁
						}else {
							//dothonig
						}
					}
				}
	
			}
		
		}catch(Exception e){
			logger.error("Some exception when get packet from cache : ",e);
			//receiveInvalidTcpPacketNum.getAndIncrement();
		}
		this.notify();
		return al;
	}
	
	

}
